package com.allwees.core.mq.rabbitmq.batch;

import com.allwees.core.mq.rabbitmq.MqException;
import com.allwees.core.mq.rabbitmq.config.RabbitMqConfig;
import lombok.extern.slf4j.Slf4j;
import org.springframework.amqp.core.AcknowledgeMode;
import org.springframework.amqp.core.MessageProperties;
import org.springframework.amqp.rabbit.connection.ConnectionFactory;
import org.springframework.amqp.rabbit.listener.BlockingQueueConsumer;
import org.springframework.amqp.rabbit.support.ActiveObjectCounter;
import org.springframework.amqp.rabbit.support.DefaultMessagePropertiesConverter;
import org.springframework.amqp.rabbit.support.MessagePropertiesConverter;
import org.springframework.beans.factory.annotation.Autowired;

import java.nio.charset.StandardCharsets;

/**
 * BatchMessageConsumer Factory
 *
 * @author yangylsky
 * @since 2019-11-25
 *
 */
@Slf4j
public class BatchMqConsumerFactory {

	@Autowired
	private RabbitMqConfig mqConfig;

	@Autowired
	private ConnectionFactory connectionFactory;

	public <M, S> BatchMqConsumer<M, S> createConsumer(String... queues) {
		return createConsumer(mqConfig.getBatchSize(), mqConfig.getBatchTimeout(), queues);
	}

	public <M, S> BatchMqConsumer<M, S> createConsumer(int batchSize, long batchTimeout, String... queues) {
		return createConsumer(null, AcknowledgeMode.AUTO, batchSize, batchTimeout, queues);
	}

	public <M, S> BatchMqConsumer<M, S> createConsumer(MessagePropertiesConverter messagePropertiesConverter, AcknowledgeMode acknowledgeMode,
			int batchSize, long batchTimeout, String... queues) {
		return createConsumer(null, acknowledgeMode, batchSize, batchTimeout, null, queues);
	}

	public <M, S> BatchMqConsumer<M, S> createConsumer(MessagePropertiesConverter messagePropertiesConverter, AcknowledgeMode acknowledgeMode,
			int batchSize, long batchTimeout, AbstractConsumer<M, S> messageConsumer, String... queues) {
		try {
			for(String queue : queues) {
				mqConfig.checkQueue(queue);
			}
			BatchMqConsumer<M, S> batchMqConsumer = new BatchMqConsumer<>(batchSize, batchTimeout, messageConsumer);

			BlockingQueueConsumer blockingQueueConsumer = new BlockingQueueConsumer(connectionFactory,
					messagePropertiesConverter == null ? defaultMessagePropertiesConverter() : messagePropertiesConverter,
					new ActiveObjectCounter<>(),
					acknowledgeMode, false, batchSize, queues);
			batchMqConsumer.setBlockingQueueConsumer(blockingQueueConsumer);

			log.info(batchMqConsumer + " is created");
			return batchMqConsumer;
		} catch(Exception e) {
			throw new MqException("Create BatchMessageConsumer failed", e);
		}
	}

	private MessagePropertiesConverter defaultMessagePropertiesConverter() {
		MessagePropertiesConverter converter = new DefaultMessagePropertiesConverter();
		MessageProperties msgProps = new MessageProperties();
		msgProps.setContentType(MessageProperties.CONTENT_TYPE_JSON);
		converter.fromMessageProperties(msgProps, StandardCharsets.UTF_8.name());
		return converter;
	}
}
